package com.yl.flink.processor;

import cn.hutool.core.date.DateUtil;
import com.yl.entity.MultiDataEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author wlf
 * @since 2022/8/26
 */
@Slf4j
public class MergeCalFunc extends ProcessWindowFunction<
        // 输入类型
        MultiDataEntity,
        // 输出类型
        MultiDataEntity,
        // 键类型
        String,
        // 窗口类型
        TimeWindow> {

    /**
     * 获取最终的最大值
     * 注意：Iterable<MultiDataEntity> 只包含一个读数 – MaxReduceFunc 计算出的预先汇总的最大值
     *
     * @param key     键类型
     * @param context 上下文
     * @param res     聚合结果
     * @param out     输出器
     */
    @Override
    public void process(String key, Context context, Iterable<MultiDataEntity> res, Collector<MultiDataEntity> out) throws Exception {
        log.info("wm:{} st:{} et:{}", DateUtil.date(context.currentWatermark()).toString(), DateUtil.date(context.window().getStart()).toString(), DateUtil.date(context.window().getEnd()).toString());
        // 输出预先汇总的最大值
        out.collect(res.iterator().next());
    }
}
